package mytest;

import homework.WordCount;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.net.URI;

/*
 *  把订单id相同的订单放在一个文件中，并且最后求出这个订单的总金额。
 *  不同的订单放在不同的文件中，并且文件以订单id来命名
 * */
public class Test_order {

	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf, "Test_order");// 设置环境参数
		job.setJarByClass(WordCount.class);// 设置整个程序的类名

		job.setMapperClass(OrderMapper.class);
		job.setReducerClass(OrderReducer.class);

		// reduce 输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		// 设置自定义的格式化输出
		job.setOutputFormatClass(MyFileOutputFormat.class);
		/*
		 * 分区数设置为4，因为只有四个订单，这里不需要重写分区函数，具体为什么可以看源码，
		 * 在这里大致意思就是每个key都有一个hash值,具体被分配到哪一个分区看hash%NumReduceTasks，
		 * 因为相同的Text的key相同，所以相同Text的key会被分到同一个分区。
		 */
		job.setNumReduceTasks(4);

		FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.3.101:9000/Test_order/input/order.txt"));
		// // 如果文件系统已存在输出文件夹则删除
		FileSystem fs = FileSystem.get(new URI("hdfs://192.168.3.101:9000"), conf, "roger");
		if (fs.exists(new Path("/Test_order/output"))) {
			fs.delete(new Path("/Test_order/output"), true);
		}

		// 注意这里是MyFileOutPutFormat，不是FileOutPutFormat，下面同样注意
		MyFileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.3.101:9000/Test_order/output"));

		/* 消除系统的默认输出文件，不过因为文件的输出格式变了，不再是FileOutPutFormat，所以这里也要改成相对应的文件输出格式 */
		LazyOutputFormat.setOutputFormatClass(job, MyFileOutputFormat.class);

		System.exit(job.waitForCompletion(true) ? 0 : 1);

	}

}

class OrderMapper extends Mapper<LongWritable, Text, Text, Text> {
	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		String[] words = value.toString().split(" ");
		context.write(new Text(words[0]), new Text(words[1] + "\t" + words[2]));
	}
}

class OrderReducer extends Reducer<Text, Text, Text, Text> {

	private MultipleOutputs<Text, Text> multipleOutputs;

	@Override
	protected void setup(Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		multipleOutputs = new MultipleOutputs<Text, Text>(context);
	}

	@Override
	protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		int sum = 0;
		for (Text value : values) {
			String[] words = value.toString().split("\t");
			sum += Integer.parseInt(words[1]);
			/*
			 * 在reduce函数中，要输出的时候，是利用定义的MultipleOutputs对象来输出
			 * multipleOutputs.write(Key输出，value输出，输出的文件名)
			 * 这里注意，如果输出的文件名中含有“/”的话，则按照这里你写的路径来输出，否则就按照FileOutPutFormat来输出，
			 * 例如你在这里输出的文件名是"/123/"+key.getNumberOne()，则输出路径就是在/123目录下，
			 * 不然的话就是在FileOutputFormat设置的outPath目录下
			 */
			multipleOutputs.write(key, value, key.toString());
		}
		/*
		 * 这里是输出统计的订单总金额信息 因为每个分区调用一个Reducer类，每个分组调用一个reduce函数，
		 * 这里其实可以看出来，因为分组是根据key来分的，所以一个分区只有一个分组，所以上面for循环执行完就代表这个分区也统计完成，
		 * 所以可以直接把这个信息写在文件末尾
		 */
		multipleOutputs.write(key, new Text(Integer.toString(sum)), key.toString());
	}

	@Override
	protected void cleanup(Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		multipleOutputs.close();
	}
}

class MyFileOutputFormat extends TextOutputFormat<Text, Text> {
	@Override
	public Path getDefaultWorkFile(TaskAttemptContext context, String extension) throws IOException {
		// TODO Auto-generated method stub
		FileOutputCommitter fileOutputCommitter = (FileOutputCommitter) getOutputCommitter(context);
		return new Path(fileOutputCommitter.getWorkPath(), getOutputName(context));
	}
}
